package com.bingxu.flink.chapter01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/6 18:32
 * @modifier :
 */

public class FlinkBoundStreamWCDemo {
    public static void main(String[] args) throws Exception {
        // 准备运行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        // 读取数据
        DataStreamSource<String> streamSource = executionEnvironment.readTextFile("D:\\Source\\JavaProject\\flinkdemo\\input\\words.txt");

        streamSource
                .flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (s, collector) -> {
                    for (String word : s.split(" ")) {
                        collector.collect(Tuple2.of(word, 1));
                    }
                })
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0)
                .sum(1)
                .print();


        // 启动
        executionEnvironment.execute();
    }
}
